package org.wmq.taskschedule.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.wmq.taskschedule.service.ThreadPoolService;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Slf4j
public class ThreadPoolServiceImpl implements ThreadPoolService {
    
    // 线程池实例
    private final ThreadPoolExecutor threadPoolExecutor;
    
    // 线程工厂，用于创建命名线程
    private static final ThreadFactory threadFactory = new ThreadFactory() {
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix = "custom-thread-pool-";
        
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, namePrefix + threadNumber.getAndIncrement());
            // 设置为非守护线程，确保任务能完成
            thread.setDaemon(false);
            // 设置线程优先级
            thread.setPriority(Thread.NORM_PRIORITY);
            return thread;
        }
    };
    
    // 线程池拒绝策略
    private static final RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            log.warn("ThreadPool is full, task will be executed by caller thread.");
            super.rejectedExecution(r, executor);
        }
    };
    
    public ThreadPoolServiceImpl() {
        // 获取处理器数量，作为核心线程数的参考
        int processorCount = Runtime.getRuntime().availableProcessors();
        
        // 配置线程池参数
        int corePoolSize = processorCount; // 核心线程数
        int maximumPoolSize = processorCount * 2; // 最大线程数
        long keepAliveTime = 60L; // 空闲线程存活时间（秒）
        TimeUnit unit = TimeUnit.SECONDS; // 时间单位
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(100); // 工作队列
        
        // 初始化线程池
        this.threadPoolExecutor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                unit,
                workQueue,
                threadFactory,
                rejectedExecutionHandler
        );
        
        log.info("Custom thread pool initialized with corePoolSize={}, maximumPoolSize={}, queueCapacity={}",
                corePoolSize, maximumPoolSize, workQueue.size());
        
        // 注册JVM关闭钩子，确保线程池在JVM退出前正确关闭
        Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
    }
    
    @Override
    public Future<?> submit(Runnable task) {
        if (task == null) {
            throw new NullPointerException("Task cannot be null");
        }
        return threadPoolExecutor.submit(task);
    }
    
    @Override
    public <T> Future<T> submit(CallableTask<T> task) {
        if (task == null) {
            throw new NullPointerException("Task cannot be null");
        }
        
        // 将自定义的CallableTask转换为Java标准的Callable
        Callable<T> callable = () -> task.call();
        return threadPoolExecutor.submit(callable);
    }
    
    @Override
    public void shutdown() {
        log.info("Shutting down thread pool...");
        threadPoolExecutor.shutdown();
        try {
            // 等待线程池关闭，最多等待60秒
            if (!threadPoolExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
                log.warn("Thread pool did not terminate within the specified time");
                // 强制关闭
                threadPoolExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            log.error("Interrupted while waiting for thread pool to shutdown", e);
            // 重新设置中断状态
            Thread.currentThread().interrupt();
            // 强制关闭
            threadPoolExecutor.shutdownNow();
        }
        log.info("Thread pool shutdown completed");
    }
    
    @Override
    public void shutdownNow() {
        log.info("Forcing immediate shutdown of thread pool...");
        threadPoolExecutor.shutdownNow();
        log.info("Immediate shutdown initiated");
    }
    
    @Override
    public int getActiveCount() {
        return threadPoolExecutor.getActiveCount();
    }
    
    @Override
    public int getQueueSize() {
        return threadPoolExecutor.getQueue().size();
    }
}